package com.pap.rocketmq.config.produce;

import com.pap.rocketmq.config.RocketMQConfig;
import com.pap.rocketmq.config.listener.PapMqTopicTransactionListener;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.TransactionMQProducer;
import org.apache.rocketmq.spring.autoconfigure.RocketMQAutoConfiguration;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.autoconfigure.AutoConfigureAfter;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import java.util.concurrent.*;

/**
 * 针对 topic 名称为：  mq-topic 的 消息生产者
 */
@Configuration
@AutoConfigureAfter(RocketMQAutoConfiguration.class)
public class PapMqTopicTransactionMQProducer {

    private static Logger logger = LoggerFactory.getLogger(PapMqTopicTransactionMQProducer.class);

    // MQ 路径
    @Autowired
    private RocketMQConfig rocketMQConfig;

    @Autowired
    private PapMqTopicTransactionListener mqTopicTransactionListener;

    @Bean("mqTopicTransactionMQProducer")
    TransactionMQProducer mqTopicTransactionMQProducer()  throws MQClientException {
        logger.info(rocketMQConfig.getPapTopicMQTopicGroup() + " : MQ事物生产者初始化开始--------------------------------------------------");

        TransactionMQProducer transactionProducer = new TransactionMQProducer(rocketMQConfig.getPapTopicMQTopicGroup());
        // Name Server 地址列表
        transactionProducer.setNamesrvAddr(rocketMQConfig.getPapRocketMQNameSrvAddr());

        //这个线程池作用就是  mqbroker端回调信息的本地处理线程池
        ExecutorService executorService = new ThreadPoolExecutor(1, 5, 100, TimeUnit.SECONDS,
                new ArrayBlockingQueue<Runnable>(2000), new ThreadFactory() {
            @Override
            public Thread newThread(Runnable r) {
                Thread thread = new Thread(r);
                thread.setName("client-transaction-msg-check-thread-mqTopic");
                return thread;
            }
        });


        transactionProducer.setExecutorService(executorService);

        transactionProducer.setTransactionListener(mqTopicTransactionListener);

        logger.info(rocketMQConfig.getPapTopicMQTopicGroup() + " : MQ事物生产者初始化结束--------------------------------------------------");

        return transactionProducer;
    }

}
